/*-
 * Copyright 2012 Guram Dukashvili
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 */
//---------------------------------------------------------------------------
#include "sys.h"
#include "str.h"
#include "idx.h"
//---------------------------------------------------------------------------
void index_copy_paste_keys(
  struct index_link * dst_link,
  uint32_t dst_position,
  struct index_link * src_link,
  uint32_t src_position,
  uintptr_t count)
{
  uintptr_t entry_size = dst_link->idx->entries_sizes[dst_link->blk->type];
  uint8_t * p = dst_link->blk->keys + dst_position * entry_size;

#if COMPILE_RUNTIME_CHECKS
  if( dst_position > dst_link->blk->count ){
    fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
    abort();
  }
  if( dst_link->blk->type != src_link->blk->type ){
    fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
    abort();
  }
  if( dst_link->blk->count + count > dst_link->idx->blk_max_keys[dst_link->blk->type] ){
    fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
    abort();
  }
#endif

  memmove(p + entry_size * count,p,(dst_link->blk->count - dst_position) * entry_size);
  memcpy(p,src_link->blk->keys + src_position * entry_size,entry_size * count);

  dst_link->blk->count += count;
}
//---------------------------------------------------------------------------
void index_cut_paste_keys(
  struct index_link * dst_link,
  uint32_t dst_position,
  struct index_link * src_link,
  uint32_t src_position,
  uintptr_t count)
{
  uintptr_t entry_size = dst_link->idx->entries_sizes[dst_link->blk->type];
  uint8_t * p = dst_link->blk->keys + dst_position * entry_size;
  uint8_t * s = src_link->blk->keys + src_position * entry_size;

#if COMPILE_RUNTIME_CHECKS
  if( src_position > src_link->blk->count ){
    fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
    abort();
  }
  if( src_position + count > src_link->blk->count ){
    fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
    abort();
  }
  if( dst_position > dst_link->blk->count ){
    fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
    abort();
  }
  if( dst_link->blk->type != src_link->blk->type ){
    fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
    abort();
  }
  //if( dst_link->blk->count + count > dst_link->idx->blk_max_keys[dst_link->blk->type] ){
  //  fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
  //  abort();
  //}
#endif

  memmove(p + entry_size * count,p,(dst_link->blk->count - dst_position) * entry_size);
  memcpy(p,s,entry_size * count);
#if COMPILE_RUNTIME_CHECKS
  memset(s,0,entry_size * count);
#endif
  memcpy(s,s + entry_size * count,(src_link->blk->count - src_position - count) * entry_size);

  src_link->blk->count -= count;
  dst_link->blk->count += count;
}
//---------------------------------------------------------------------------
static __inline__ intptr_t index_blocks_map_compare2(
  const struct byte_size_int * p1,
  const struct index_block_mapper_entry * p2,
  struct index * idx)
{
  return memcmp(p1,&p2->block_index,idx->block_index_size);
}
//---------------------------------------------------------------------------
void index_read_block(struct index * idx,struct index_link * link)
{
  enum INDEX_FILE_TYPE ft = INDEX_IDX_FILE_HANDLE;
  uintptr_t jrn_file_index = 0;
  portable_off_t off;
  int file_handle;
  portable_ssize_t r;

  {
    uintptr_t insert_position;
    intptr_t p;
    
    rwlock_acquire_read_lock(&idx->blocks_map_lock);

    p = array_lookupf(&idx->blocks_map,&link->block_index,&insert_position,index_blocks_map_compare2);
    if( p >= 0 ){
      struct index_block_mapper_entry * e = array_get(&idx->blocks_map,p);
      off = (portable_off_t) bsint_get_be64toh(&e->jrn_block_index,idx->block_index_size);
      jrn_file_index = e->jrn_file_index;
      ft = INDEX_JRN_FILE_HANDLE;
    }
    else {
      off = (portable_off_t) bsint_get_be64toh(&link->block_index,link->idx->block_index_size);
    }

    rwlock_release_read_lock(&idx->blocks_map_lock);
  }

  file_handle = index_get_file_handle(idx,ft,jrn_file_index,true);

  r = portable_pread(file_handle,link->blk,link->idx->cache->block_size,off * idx->cache->block_size);
  if( r < 0 ){
    perror(NULL);
#if COMPILE_RUNTIME_CHECKS
    fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
#endif
    abort();
  }

  index_put_file_handle(idx,ft,jrn_file_index,file_handle);

  memset(link->blk,0,link->idx->cache->block_size - r);

  if( !index_is_block_index_extent(idx,&link->block_index) )
    link->blk->count = be32toh(link->blk->count);
}
//---------------------------------------------------------------------------
void index_write_block(struct index * idx,struct index_link * link)
{
  int file_handle;
  portable_off_t off;
  portable_ssize_t w;

  if( link == NULL ) return;

  file_handle = index_get_file_handle(idx,INDEX_IDX_FILE_HANDLE,0,true);

  off = (portable_off_t) bsint_get_be64toh(&link->block_index,idx->block_index_size);

  w = portable_pwrite(file_handle,link->blk,idx->cache->block_size,off * idx->cache->block_size);

  if( w < 0 ){
    perror(NULL);
#if COMPILE_RUNTIME_CHECKS
    fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
#endif
    abort();
  }

  index_put_file_handle(idx,INDEX_IDX_FILE_HANDLE,0,file_handle);
}
//---------------------------------------------------------------------------
void index_shrink_cache_lru(struct index * idx,bool watermark)
{
  struct shared_cache * cache = idx->cache;
  uintptr_t low_watermark = cache->max_size * cache->low_watermark;

  mutex_acquire(&cache->links_mutex);

  for(;;){
    struct list_node * node;
    struct index_link * lnk;

    if( cache->links_lru.count == 0 ) break;

    node = cache->links_lru.tail;
    lnk = list_node2data(&cache->links_lru,node);

#if COMPILE_RUNTIME_CHECKS
    if( lnk->list != &cache->links_lru ){
      fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
      abort();
    }
#endif

    list_remove(&cache->links_lru,lnk);
    hash_remove(&cache->links_hash,lnk);

    index_link_delete(lnk);

    cache->cur_size -= cache->block_size;
    if( watermark && cache->cur_size < low_watermark ) break;
  }

  mutex_release(&cache->links_mutex);
}
//---------------------------------------------------------------------------
void index_shrink_cache(struct index * idx)
{
  struct shared_cache * cache = idx->cache;
  uintptr_t high_watermark = cache->max_size * cache->high_watermark;

  if( cache->cur_size <= high_watermark ) return;

#ifndef NDEBUG
  //fprintf(stderr,"before shrink cache: %" PRIuPTR "\n",cur_cache_size);
#endif

  index_shrink_cache_lru(idx,true);
}
//---------------------------------------------------------------------------
void index_link_block_n(struct index * idx,struct index_link * link)
{
  struct shared_cache * cache = idx->cache;

  mutex_acquire(&cache->links_mutex);
  
  list_insert_head(&cache->links_linked,link);
  link->list = &cache->links_linked;
  hash_insert(&cache->links_hash,link);
#if COMPILE_RUNTIME_CHECKS
  if( errno != 0 ){
    fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
    abort();
  }
#endif
  cache->cur_size += cache->block_size;
  
  mutex_release(&cache->links_mutex);
}
//---------------------------------------------------------------------------
struct index_link * index_link_block_v(struct index * idx,const byte_size_int * block_index,bool read)
{
  struct shared_cache * cache = idx->cache;
  struct index_link * link, * slnk = alloca(sizeof(struct index_link));

  memcpy(&slnk->block_index,block_index,idx->block_index_size);
  slnk->idx = idx;

  mutex_acquire(&cache->links_mutex);

  link = hash_lookup(&cache->links_hash,slnk);

  if( link != slnk ){
#if COMPILE_RUNTIME_CHECKS
    if( link->idx != idx ){
      fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
      abort();
    }
#endif

    if( link->list == &cache->links_lru ){
      list_remove(&cache->links_lru,link);
      list_insert_head(&cache->links_linked,link);
      link->list = &cache->links_linked;
    }

    link->linked++;

    if( link->io_inprogress ){
      rwlock_acquire_read_lock(&link->lock); // wait IO completion
#if COMPILE_RUNTIME_CHECKS
      if( link->io_inprogress ){
        fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
        abort();
      }
#endif
      // now IO complete
      rwlock_release_read_lock(&link->lock);
    }

    cache->hit_count++;

    mutex_release(&cache->links_mutex);
  }
  else {
    int64_t t;

    link = index_link_new_v(cache->block_size);
    link->linked = 1;
    link->idx = idx;

    cache->cur_size += cache->block_size;

    memcpy(&link->block_index,block_index,idx->block_index_size);

    rwlock_acquire_write_lock(&link->lock);

    list_insert_head(&cache->links_linked,link);
    link->list = &cache->links_linked;
    hash_insert(&cache->links_hash,link);

    link->io_inprogress = true;

    mutex_release(&cache->links_mutex);

    t = get_time();

    if( read )
      index_read_block(idx,link);
    else
      memset(link->blk,0,cache->block_size);
    
    t = get_time() - t;

    link->io_inprogress = false;
    rwlock_release_write_lock(&link->lock);

    mutex_acquire(&cache->links_mutex);
    cache->iowait_time += t;
    cache->miss_count++;
    mutex_release(&cache->links_mutex);
  }
  return link;
}
//---------------------------------------------------------------------------
void index_unlock_stack(struct index * idx,struct index_btree_stack * stack)
{
  intptr_t sp, link;

  idx = idx;
  for( sp = stack->sp; sp >= 0; sp-- ){
    struct index_btree_stack_entry * e = array_get(&stack->entries,sp);

    for( link = sizeof(e->links) / sizeof(e->links[0]) - 1; link >= 0; link-- ){
      struct index_link * plnk = e->links[link];

      if( plnk == NULL ) continue;

      rwlock_release(&plnk->lock,e->links_locks[link]);
      e->links_locks[link] = RWL_NONE;
    }
  }
}
//---------------------------------------------------------------------------
void index_unlink_block(struct index * idx,struct index_link * link,enum RWLOCK_TYPE link_lock,enum RWLOCK_TYPE lock)
{
  struct index_link_change change;
  uintptr_t insert_position;
  intptr_t i;

  rwlock_acquire(&idx->changed_links_lock,lock);

  change.link = link;
  i = array_lookup(&idx->changed_links,&change,&insert_position);

  mutex_acquire(&idx->cache->links_mutex);
  // unlink only if block is not have been changed (stay in memory)
  // else will be unlinked after writing to journal in index_write_journal
  if( --link->linked == 0 && i < 0 ){
    list_remove(&idx->cache->links_linked,link);
    list_insert_head(&idx->cache->links_lru,link);
    change.link->list = &idx->cache->links_lru;
  }

  mutex_release(&idx->cache->links_mutex);

  rwlock_release(&link->lock,link_lock);
}
//---------------------------------------------------------------------------
void index_unwind_stack(struct index * idx,struct index_btree_stack * stack,enum RWLOCK_TYPE lock)
{
  struct shared_cache * cache = idx->cache;
  uintptr_t insert_position, mod;
  intptr_t sp, link, i;
  bool optimize = false;

  rwlock_acquire(&idx->changed_links_lock,lock);

  for( sp = stack->sp; sp >= 0; sp-- ){
    struct index_btree_stack_entry * e = array_get(&stack->entries,sp);

    for( link = sizeof(e->links) / sizeof(e->links[0]) - 1; link >= 0; link-- ){

      struct index_link_change change, * p;
      struct index_link * plnk = e->links[link];

      if( plnk == NULL ) continue;

      if( plnk->blk->count > idx->blk_max_keys[plnk->blk->type]
          && plnk->blk->count * idx->entries_sizes[plnk->blk->type] > 64 * 1024 )
        idx->optimize = optimize = true;

      change.what = e->links_change[link];
      change.link = plnk;

      mod = 0;

      i = array_lookup(&idx->changed_links,&change,&insert_position);

      if( i < 0 ){
        if( change.what != ILC_NONE ){
          i = insert_position;
          array_insert_copy(&idx->changed_links,insert_position,&change);
          mod = cache->block_size;
        }
      }
      else if( change.what != ILC_NONE ){
        if( change.what != ILC_UPD ){
          p = array_get(&idx->changed_links,i);
          p->what = change.what;
        }
      }

      if( change.what == ILC_DEL )
        index_block_release(idx,&plnk->block_index);

      //e->links_change[link] = ILC_NONE;
      //e->links[link] = NULL;

      mutex_acquire(&cache->links_mutex);
#if COMPILE_RUNTIME_CHECKS
      if( plnk->list != &cache->links_linked ){
        fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
        abort();
      }
#endif
      // unlink only if block is not have been changed (stay in memory)
      // else will be unlinked after writing to journal in index_write_journal
      if( --plnk->linked == 0 && i < 0 ){
        list_remove(&cache->links_linked,plnk);
        list_insert_head(&cache->links_lru,plnk);
        change.link->list = &cache->links_lru;
      }

      mutex_release(&cache->links_mutex);

      rwlock_release(&plnk->lock,e->links_locks[link]);
    }
  }

  rwlock_release(&idx->changed_links_lock,lock);

  index_shrink_cache(idx);

  if( idx->changed_links.count > 0 || optimize ){
    if( !thread_active(&idx->write_journal_thread) )
      thread_start(&idx->write_journal_thread);
    if( optimize || cache->block_size * idx->changed_links.count > cache->max_size * cache->flush_watermark )
      semaphore_post(&idx->write_journal_wait);
  }
}
//---------------------------------------------------------------------------
static __inline__ void index_stack_entry_lookup_key(
  struct index_btree_stack_entry * e,
  const void * key)
{
  struct index_link * link = e->link;
  intptr_t low = 0, high = (intptr_t) link->blk->count - 1;
  uintptr_t entry_size = link->idx->entries_sizes[link->blk->type];
  uintptr_t key_size = link->idx->key_size;
  uint8_t * keys = link->blk->keys;

  e->c = -1;
  e->position = 0;

  for(;;){

    uintptr_t position = (low + high) / 2;

    if( low > high ) break;

    e->position = position;
    e->c = memcmp(key,keys + position * entry_size,key_size);

    if( e->c > 0 ){
      low = position + 1;
    }
    else if( e->c < 0 ){
      high = position - 1;
    }
    else
      break;
  }
}
//---------------------------------------------------------------------------
void index_lookup_key(struct index_btree_stack_entry * e,const void * key,intptr_t direction)
{
  if( direction < 0 ){
    e->position = e->c = 0;
    //TODO:
  }
  else if( direction > 0 ){
    e->position = e->link->blk->count - 1;
    e->c = 0;
    //TODO:
  }
  else {
    index_stack_entry_lookup_key(e,key);
  }
  e->insert_position = e->position + (e->c > 0);
  e->next_level_position = e->position - (e->c < 0);
  e->next_level_position += e->next_level_position < 0;
}
//---------------------------------------------------------------------------
void index_get_key_path(
  struct index * idx,
  struct index_btree_stack * stack,
  const void * key,
  intptr_t direction)
{
  struct index_btree_stack_entry * e;
  byte_size_int * block_index = alloca(idx->block_index_size);
  memset(block_index,0,idx->block_index_size);

  stack->sp = 0;

  for(;;){
    if( (uintptr_t) stack->sp == stack->entries.count )
      array_append(&stack->entries);

    e = array_get(&stack->entries,stack->sp);
    index_btree_stack_entry_init(e);
    e->link = index_link_block(idx,block_index);
    if( errno != 0 ) break;
    if( e->link->blk->type != INDEX_NODE && e->link->blk->type != INDEX_LEAF ){
      errno = EINVAL;
      return;
    }

    rwlock_acquire_read_lock(&e->link->lock);
    e->link_lock = RWL_READ;

    index_lookup_key(e,key,direction);

    if( direction != 0 ) e->c = direction;

    if( e->link->blk->type == INDEX_LEAF ) break;

#if COMPILE_RUNTIME_CHECKS
    if( e->link->blk->type != INDEX_NODE ){
      fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
      abort();
    }
#endif

    // check next level
    memcpy(block_index,
      index_block_link(e->link,e->next_level_position),
      idx->block_index_size);

    stack->sp++;
  }
#ifndef NDEBUG
  if( stack->sp > idx->max_sp )
    idx->max_sp = stack->sp;
#endif
}
//---------------------------------------------------------------------------
struct select_key_cb {
  struct index * idx;
  struct thread * thr;
  const void * key;
  void * data;
  int err;
};
//---------------------------------------------------------------------------
static void index_lookup_helper(
  struct index * idx,
  struct index_btree_stack * stack,
  const void * key,
  void * data)
{
  int err = ENOENT;
  struct index_btree_stack_entry * e;

  index_get_key_path(idx,stack,key,0);

  e = array_get(&stack->entries,stack->sp);

  if( errno != 0 ){
    err = errno;
  }
  else if( e->c == 0 ){
    if( data != NULL )
      memcpy(data,index_block_entry(e->link,e->position) + idx->key_size,idx->data_size);
    err = 0;
  }

  index_unwind_stack(idx,stack,RWL_READ);

  index_btree_stack_destroy(stack);

  errno = err;
}
//---------------------------------------------------------------------------
static void index_lookup_entry(struct thread * thr,struct select_key_cb * cb)
{
  struct index_btree_stack stack;

  thr = thr;
  index_btree_stack_init(&stack);

  index_lookup_helper(cb->idx,&stack,cb->key,cb->data);
  cb->err = errno;

  index_btree_stack_destroy(&stack);
}
//---------------------------------------------------------------------------
void index_lookup(
  struct index * idx,
  struct index_btree_stack * stack,
  const void * key,
  void * data,
  struct index ** p_mvcc_idx)
{
  int err;
  //intptr_t i;
  //bool mvcc_inprogress = false;
  //struct select_key_cb * cb = NULL, * result = NULL;
  //array * cbs = alloca(sizeof(array) + sizeof(struct select_key_cb) * idx->mvcc_pool_size);

  //array_init(cbs);
  //array_set_element_size(cbs,sizeof(struct select_key_cb),sizeof(struct select_key_cb) * idx->mvcc_pool_size);

  //semaphore_acquire(&idx->mvcc_lock);

  //array_resize(cbs,i = idx->mvcc_indexes_pool.pointers.count);

  //for( i = i - 1; i >= 0; i-- ){
  //  cb = array_get(cbs,i);
  //  cb->idx = vector_get(&idx->mvcc_indexes_pool,i);
  //  cb->idx->mvcc_ref_count++;
  //  cb->key = key;
  //  cb->data = data;
  //  cb->err = -1;
  //  cb->thr = NULL;
  //}

  //semaphore_release(&idx->mvcc_lock);

  //for( i = cbs->count - 1; i >= 0; i-- ){
  //  cb = array_get(cbs,i);
  //  cb->thr = thread_pool_spare(&idx->mvcc_thread_pool,index_lookup_entry,cb,0);
  //}

  index_lookup_helper(idx,stack,key,data);
  err = errno;
  if( p_mvcc_idx != NULL ) *p_mvcc_idx = idx;

  //for( i = cbs->count - 1; i >= 0; i-- ){
  //  cb = array_get(cbs,i);
  //  thread_wait(cb->thr);
  //  if( cb->err == 0 ) result = cb;
  //}

  //semaphore_acquire(&idx->mvcc_lock);
  //for( i = cbs->count - 1; i >= 0; i-- ){
  //  cb = array_get(cbs,i);
  //  if( p_mvcc_idx != NULL && cb == result ){
  //    mvcc_inprogress = cb->idx->mvcc_inprogress;
  //    if( mvcc_inprogress )
  //      cb->idx->mvcc_waiters++;
  //    else
  //      cb->idx->mvcc_inprogress = true;
  //    *p_mvcc_idx = cb->idx;
  //  }
  //  else
  //    cb->idx->mvcc_ref_count--;
  //}
  //semaphore_release(&idx->mvcc_lock);

  //if( result != NULL ) err = result->err;

  //array_destroy(cbs);

  //if( p_mvcc_idx != NULL && err == 0 && mvcc_inprogress )
  //  semaphore_wait(&(*p_mvcc_idx)->mvcc_wait);

  errno = err;
}
//---------------------------------------------------------------------------
void index_update_key(
  struct index * idx,
  struct index_btree_stack * stack,
  const void * key,
  const void * data)
{
  int err = ENOENT;
  struct index_btree_stack_entry * e;

  index_get_key_path(idx,stack,key,0);

  e = array_get(&stack->entries,stack->sp);

  if( errno != 0 ){
    err = errno;
  }
  else if( e->c == 0 ){
    rwlock_acquire_read_lock(&idx->changed_links_lock);

    if( data != NULL ){
      rwlock_upgrade(&e->link->lock);
      e->link_lock = RWL_WRITE;

      memcpy(index_block_entry(e->link,e->position) + idx->key_size,data,idx->data_size);
      e->link_change = ILC_UPD;
    }

    err = 0;

    rwlock_upgrade(&idx->changed_links_lock);
    index_unwind_stack(idx,stack,RWL_NONE);
    rwlock_release_write_lock(&idx->changed_links_lock);
  }
  errno = err;
}
//---------------------------------------------------------------------------
struct index * index_get_mvcc(struct index * idx)
{
  intptr_t i;
  uintmax_t count;
  struct index * mvcc_idx = NULL;

  semaphore_acquire(&idx->mvcc_lock);

  for(;;){
    mvcc_idx = NULL;

    count = ~(uintmax_t) 0;
    for( i = idx->mvcc_indexes_pool.pointers.count - 1; i >= 0; i-- ){
      struct index * p = vector_get(&idx->mvcc_indexes_pool,i);
      if( !p->mvcc_inprogress && p->mvcc_count < count ){
        mvcc_idx = p;
        count = p->mvcc_count;
      }
    }

    if( mvcc_idx == NULL && idx->mvcc_indexes_pool.pointers.count < idx->mvcc_pool_size ){
      mvcc_idx = vector_append_get(&idx->mvcc_indexes_pool);
      mvcc_idx->id = idx->id;
      mvcc_idx->mvcc_id = idx->mvcc_indexes_pool.pointers.count;
      string_copy(&mvcc_idx->data_dir,&idx->data_dir);
      mvcc_idx->key_size = idx->key_size;
      mvcc_idx->data_size = idx->data_size;
      mvcc_idx->block_index_size = idx->block_index_size;
      memcpy(mvcc_idx->blk_max_keys,idx->blk_max_keys,sizeof(mvcc_idx->blk_max_keys));
      memcpy(mvcc_idx->entries_sizes,idx->entries_sizes,sizeof(mvcc_idx->entries_sizes));
    }

    if( mvcc_idx != NULL ) break;

    idx->mvcc_waiters++;

    semaphore_release(&idx->mvcc_lock);
    semaphore_wait(&idx->mvcc_wait);

    idx->mvcc_waiters--;
  }
  mvcc_idx->mvcc_ref_count++;
  mvcc_idx->mvcc_count++;
  mvcc_idx->mvcc_inprogress = true;

  semaphore_release(&idx->mvcc_lock);

  return mvcc_idx;
}
//---------------------------------------------------------------------------
void index_put_mvcc(struct index * idx,struct index * mvcc_idx)
{
  semaphore_acquire(&idx->mvcc_lock);

  mvcc_idx->mvcc_ref_count--;

  if( mvcc_idx->mvcc_waiters == 0 )
    mvcc_idx->mvcc_inprogress = false;
  else {
    mvcc_idx->mvcc_waiters--;
    semaphore_post(&mvcc_idx->mvcc_wait);
  }

  if( idx->mvcc_waiters > 0 )
    semaphore_post(&idx->mvcc_wait);
  else
    semaphore_release(&idx->mvcc_lock);
}
//---------------------------------------------------------------------------
#ifndef NDEBUG
//---------------------------------------------------------------------------
void DEBUG_index_write_changed_blocks(struct index * idx)
{
  uintptr_t i;

  for( i = 0; i < idx->changed_links.count; i++ ){
    struct index_link_change * changed_link = array_get(&idx->changed_links,i);
    index_write_block(idx,changed_link->link);
  }
}
//---------------------------------------------------------------------------
void DEBUG_index_dump_stack_entry(FILE * stream,struct index_btree_stack_entry * e)
{
  uintptr_t i;
  const uint8_t * entry = index_block_entry(e->link,0);

  fprintf(stream,"block: ");
  bsint_print(stream,&e->link->block_index,e->link->idx->block_index_size);
  fprintf(stream," [");
  for( i = 0; i < e->link->idx->key_size; i++ )
    fprintf(stream,i == 0 ? "%02X" : ".%02X",entry[i]);
  entry = index_block_entry(e->link,e->link->blk->count - 1);
  fprintf(stream,",");
  for( i = 0; i < e->link->idx->key_size; i++ )
    fprintf(stream,i == 0 ? "%02X" : ".%02X",entry[i]);
  fprintf(stream,"]\n\tc: %" PRIdPTR ", position: %" PRIXPTR ", next position: %" PRIXPTR "\n",e->c,e->position,e->next_level_position);
  if( e->link->blk->type == INDEX_NODE ){
    const uint8_t * entry = index_block_entry(e->link,e->next_level_position);
    const byte_size_int * block_index = index_block_link(e->link,e->next_level_position);
    fprintf(stream,"\tkey: ");
    for( i = 0; i < e->link->idx->key_size; i++ )
      fprintf(stream,i == 0 ? "%02X" : ".%02X",entry[i]);
    fprintf(stream,", next block: ");
    for( i = 0; i < e->link->idx->block_index_size; i++ )
      fprintf(stream,"%02X",((const uint8_t *) block_index)[i]);
    fprintf(stream,"\n");
  }
  else {
    //DEBUG_index_dump_block_keys(stream,e->xlink);
  }
}
//---------------------------------------------------------------------------
void DEBUG_index_dump_block_keys_range(FILE * stream,struct index_link * link)
{
  uintptr_t i;
  const uint8_t * entry = index_block_entry(link,0);

  fprintf(stream,"block: ");
  bsint_print(stream,&link->block_index,link->idx->block_index_size);
  fprintf(stream,", prev: ");
  bsint_print(stream,&link->blk->prev_block_index,link->idx->block_index_size);
  fprintf(stream," [");
  for( i = 0; i < link->idx->key_size; i++ )
    fprintf(stream,i == 0 ? "%02X" : ".%02X",entry[i]);
  entry = index_block_entry(link,link->blk->count - 1);
  fprintf(stream,",");
  for( i = 0; i < link->idx->key_size; i++ )
    fprintf(stream,i == 0 ? "%02X" : ".%02X",entry[i]);
  fprintf(stream,"]\n");
}
//---------------------------------------------------------------------------
void DEBUG_index_dump_block_keys(FILE * stream,struct index_link * link)
{
  uint8_t * keys;
  uintptr_t i, j;

  if( link == NULL ) return;

  fprintf(stream,"block keys dump --> ");
  bsint_print(stream,&link->block_index,link->idx->block_index_size);
  fprintf(stream,"\n");

  keys = link->blk->keys;

  for( i = 0; i < link->blk->count; i++ ){
    fprintf(stream,"%4" PRIXPTR " ",i);
    for( j = 0; j < link->idx->key_size; j++ )
      fprintf(stream,j == 0 ? "%02X" : ".%02X",(unsigned) keys[j]);
    if( link->blk->type == INDEX_NODE ){
      fprintf(stderr," ");
      for( j = 0; j < link->idx->block_index_size; j++ )
        fprintf(stream,j == 0 ? "%02X" : ".%02X",(unsigned) keys[link->idx->key_size + j]);
    }
    fprintf(stream,"\n");
    keys += link->idx->entries_sizes[link->blk->type];
  }
}
//---------------------------------------------------------------------------
intptr_t DEBUG_index_lookup_key(
  struct index * idx,
  struct index_link * root,
  const byte_size_int * block_index,
  const void * key)
{
  struct index_btree_stack_entry e;

  if( bsint_is_zero(block_index,idx->block_index_size) ){
    e.link = root;
  }
  else {
    e.link = index_link_block(idx,block_index);
    if( errno != 0 ) return -1;
  }

  index_lookup_key(&e,key,0);

  if( e.c != 0 || e.link->blk->type == INDEX_NODE ){
    e.c = DEBUG_index_lookup_key(idx,root,
      (const byte_size_int *) index_block_link(e.link,e.next_level_position),key);
  }

  if( e.link != root ){
    mutex_acquire(&idx->cache->links_mutex);
    e.link->linked--;
    mutex_release(&idx->cache->links_mutex);
  }

  return e.c;
}
//---------------------------------------------------------------------------
intptr_t DEBUG_index_lookup_key_dump(
  struct index * idx,
  struct index_link * root,
  const byte_size_int * block_index,
  const void * key)
{
  struct index_btree_stack_entry e;

  if( bsint_is_zero(block_index,idx->block_index_size) ){
    e.link = root;
  }
  else {
    e.link = index_link_block(idx,block_index);
    if( errno != 0 ) return -1;
  }

  index_lookup_key(&e,key,0);
  DEBUG_index_dump_stack_entry(stderr,&e);

  if( e.c != 0 || e.link->blk->type == INDEX_NODE ){
    e.c = DEBUG_index_lookup_key_dump(idx,root,
      (const byte_size_int *) index_block_link(e.link,e.next_level_position),key);
  }

  if( e.link != root ){
    mutex_acquire(&idx->cache->links_mutex);
    e.link->linked--;
    mutex_release(&idx->cache->links_mutex);
  }

  return e.c;
}
//---------------------------------------------------------------------------
#endif
//---------------------------------------------------------------------------
